package com.diao.flink.windowAndTimer;

import jodd.datetime.TimeUtil;
import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.util.concurrent.TimeUnit;

/**
 * @author: chenzhidiao
 * @date: 2021/2/3 08:08
 * @description: 通过自定义输入源来进行消息处理
 * 程序启动后，第13秒发送2条消息，第16秒发送1条消息
 * 每隔5秒统计最近10秒每隔单出现次数
 * 第1，2条消息在第一个窗口
 * 第1，2，3条消息在第二个窗口
 * 第3条消息在第三个窗口
 *
 * 结果：
 * 当前时间：16:53:20，过13秒之后是16：53：33
 * 当前系统时间：16:53:35
 * 窗口处理时间：16:53:35
 * 窗口开始时间：16:53:25
 * 窗口结束时间：16:53:35
 * =====================================================
 * (flink,2)
 * 当前系统时间：16:53:40
 * 窗口处理时间：16:53:40
 * 窗口开始时间：16:53:30
 * 窗口结束时间：16:53:40
 * =====================================================
 * (flink,3)
 * 当前系统时间：16:53:45
 * 窗口处理时间：16:53:45
 * 窗口开始时间：16:53:35
 * 窗口结束时间：16:53:45
 * =====================================================
 * (flink,1)
 *
 * Process finished with exit code 130 (interrupted by signal 2: SIGINT)
 */
public class WindowWordCountBySource {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);

        DataStreamSource<String> streamSource = env.addSource(new TestSource());

        streamSource.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {

            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = s.split(",");
                for (String word : words) {
                    collector.collect(Tuple2.of(word,1));
                }
            }
        })
        .keyBy(tuple ->tuple.f0)
        .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
        .process(new SumProcessFunction())
                .print().setParallelism(1);

        env.execute("wordCountWithWindow");


    }
}

class TestSource implements SourceFunction<String>{



    FastDateFormat dateformat = FastDateFormat.getInstance("HH:mm:ss");

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        String currentTime = String.valueOf(System.currentTimeMillis());

        System.out.println(currentTime.substring(currentTime.length()-4));

        while(Integer.valueOf(currentTime.substring(currentTime.length()-4))>100){
            currentTime = String.valueOf(System.currentTimeMillis());
            continue;

        }
        System.out.println("当前时间："+dateformat.format(System.currentTimeMillis()));

        TimeUnit.SECONDS.sleep(13);

        String log = "flink";
        //13秒发送2条数据
        sourceContext.collect(log);
        sourceContext.collect(log);

        //16秒发送1条数据
        TimeUnit.SECONDS.sleep(3);
        sourceContext.collect(log);


        TimeUnit.SECONDS.sleep(3000000);

    }

    @Override
    public void cancel() {

    }
}
class SumProcessFunction extends ProcessWindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>, String, TimeWindow> {

    FastDateFormat dateformat = FastDateFormat.getInstance("HH:mm:ss");

    @Override
    public void process(String key,  Context context,  Iterable<Tuple2<String, Integer>> elements,  Collector<Tuple2<String, Integer>> out) throws Exception {


        System.out.println("当前系统时间："+dateformat.format(System.currentTimeMillis()));
        System.out.println("窗口处理时间："+dateformat.format(context.currentProcessingTime()));

        System.out.println("窗口开始时间："+dateformat.format(context.window().getStart()));
        System.out.println("窗口结束时间："+dateformat.format(context.window().getEnd()));

        System.out.println("=====================================================");
        int count = 0;
        for(Tuple2<String,Integer> e :elements){
            //统计单词个数
            count++;
        }
        out.collect(Tuple2.of(key,count));

    }
}